package com.galeno.day09.test;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.OutputTag;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2623:06
 */
public class GetLateData {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",12345);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);

        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                long time = Long.parseLong(split[0]);
                String word = split[1];
                int count = Integer.parseInt(split[2]);

                return Tuple3.of(time, word, count);
            }
        }).setParallelism(1);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> wordAndMark = tpStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> element) {
                return element.f0;

            }
        });
        KeyedStream<Tuple3<Long, String, Integer>, String> keyedStream = wordAndMark.keyBy(x -> x.f1);

        OutputTag<Tuple3<Long, String, Integer>> latedate = new OutputTag<Tuple3<Long, String, Integer>>("latedate"){};

        WindowedStream<Tuple3<Long, String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sideOutputLateData(latedate);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> sumed = windowedStream.sum(2);

        DataStream<Tuple3<Long, String, Integer>> lateStream = sumed.getSideOutput(latedate);
        lateStream.print("late-date");
        sumed.print("普通");
        env.execute();



    }
}
